package com.sui.bigdata.rtcadmin.async;

import com.alibaba.fastjson.JSONObject;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.HttpUtil;
import com.google.common.collect.Lists;
import com.sui.bigdata.rtcadmin.repository.model.JobConfig;
import com.sui.bigdata.rtcadmin.util.RedisUtils;
import com.sui.bigdata.rtcadmin.util.SendMsgUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_PREFIX;
import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_SUFFIX;

/**
 * @author YongChen
 * @date 2019/12/12 19:04
 * @description
 * @email yong_chen@sui.com
 */
public class BackpressureAlarm implements Runnable {

    protected static final Logger logger = LoggerFactory.getLogger(BackpressureAlarm.class);

    private static String ID = "backpressure";


    private final Long ttl = 24 * 60 * 60L;

    private JobConfig jobConfig;
    private String alarmUrl;
    private String backPressureUrl;
    private String description;
    private String vecId;
    private RedisUtils redisUtils;
    private Integer todayMaxAlarmTimes;


    public BackpressureAlarm(JobConfig jobConfig,
                             String alarmUrl,
                             String backPressureUrl,
                             String description,
                             String vecId,
                             RedisUtils redisUtils,
                             Integer todayMaxAlarmTimes) {
        this.jobConfig = jobConfig;
        this.alarmUrl = alarmUrl;
        this.backPressureUrl = backPressureUrl;
        this.description = description;
        this.vecId = vecId;
        this.redisUtils = redisUtils;
        this.todayMaxAlarmTimes = todayMaxAlarmTimes;


    }

    @Override
    public void run() {
        String alarmUrlNew = alarmUrl;
        if (StringUtils.isNotBlank(jobConfig.getAlarmGroupPhone())) {
            alarmUrlNew = alarmUrl + jobConfig.getAlarmGroupPhone();
        }
        StringResponse response = null;
        try {
            for (int i = 0; i < 3; i++) {
                try {
                    HttpGet httpGet = new HttpGet(backPressureUrl);
                    RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(60000).build();
                    httpGet.setConfig(requestConfig);
                    response = HttpUtil.get(httpGet);
                    break;
                } catch (Exception e) {
                    if (i == 2) {
                        throw e;
                    }
                    Thread.sleep(200);
                }
            }
            if (null == response) {
                return;
            }
            JSONObject jsonobj = JSONObject.parseObject(response.getResponseBody());
            for (Object node : jsonobj.getJSONArray("subtasks")) {
                JSONObject jsonNode = (JSONObject) node;
                String level = jsonNode.get("backpressure-level").toString();
                String ratio = jsonNode.get("ratio").toString();

                SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
                String todayDate = formatter.format(new Date(System.currentTimeMillis()));

                String key = ID + jobConfig.getJobName() + vecId + todayDate;
                Integer todaySendTimes = redisUtils.getInteger(key);

                if ("high".equals(level) && todaySendTimes % 2 == 0 && todaySendTimes < todayMaxAlarmTimes * 2) {
                    SendMsgUtils.sendBackpressureMsg(alarmUrlNew, jobConfig.getJobName(), jobConfig.getDeveloper(), level, ratio, description, CustomConfiguration.getString(FLINK_WEB_PREFIX) + jobConfig.getAppId() + FLINK_WEB_SUFFIX);
                    redisUtils.set(key, todaySendTimes + 1, ttl);
                }
                if ("ok".equals(level) && todaySendTimes % 2 == 1) {
                    SendMsgUtils.sendBackpressureMsg(alarmUrlNew, jobConfig.getJobName(), level, ratio, description);
                    redisUtils.set(key, todaySendTimes + 1, ttl);
                    logger.warn("the subtask of {} called {} , backPressure  level:{} ratio:{}.", jobConfig.getJobName(), description, level, ratio);
                }
            }
        } catch (Exception e) {
            logger.warn("AlarmBackpresureTask重试2次后执行失败: ", e);
        }
    }

    public static List<BackpressureAlarm> build(JobConfig jobConfig, String alarmUrl, RedisUtils redisUtils, Integer todayMaxAlarmTimes) throws Exception {
        List<BackpressureAlarm> backpressureAlarms = Lists.newArrayList();
        String webUrl = jobConfig.getWebUrl();
        String planUrl = webUrl + "/jobs/" + jobConfig.getJobId() + "/plan";
        StringResponse response = null;
        for (int i = 0; i < 3; i++) {
            try {
                HttpGet httpGet = new HttpGet(planUrl);
                RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(60000).build();
                httpGet.setConfig(requestConfig);
                response = HttpUtil.get(httpGet);
                break;
            } catch (Exception e) {
                if (i == 2) {
                    logger.warn("AlarmBackpresure {} 重试2次后执行失败 {}", jobConfig.getJobName(), e);
                    return null;
                }
                Thread.sleep(200);
            }
        }
        if (null == response) {
            return null;
        }
        JSONObject plan = JSONObject.parseObject(response.getResponseBody());
        for (Object node : plan.getJSONObject("plan").getJSONArray("nodes")) {
            JSONObject jsonNode = (JSONObject) node;
            String vecId = jsonNode.get("id").toString();
            String description = getDescription(jsonNode.get("description").toString());
            String backpressureUrl = webUrl + "/jobs/" + jobConfig.getJobId() + "/vertices/" + vecId + "/backpressure";
            backpressureAlarms.add(new BackpressureAlarm(jobConfig, alarmUrl, backpressureUrl, description, vecId, redisUtils, todayMaxAlarmTimes));
        }
        return backpressureAlarms;
    }

    public static String getDescription(String description) {
        int length = description.length();
        if (length >= 40) {
            return description.substring(0, 40);
        } else {
            return description;
        }
    }


}
